package com.cctc.juc.demo.Bitc.Fjuctools.DPhaser;

import java.util.concurrent.*;

public class PhaserDemo {

    /**
     * 线程间通信方式之使用 Phaser 移相器：多个任务线程之间需要相互等待，直到达到某个 [阶段]，才能继续执行后续的操作。
     * Phaser [移相器] 是一个常用的线程同步工具，它可以将任务线程的运行分为多个阶段，先到达前一阶段（phase 0）的任务线程需要自我等待（await），全部到达后再各自前进到（advance）下一个阶段（phase 1），依次类推，直到终点。
     * 名词解释：
     * - party & parties：参与者 & 数量，所有需要同步的参与者必须持有同一个 phaser 对象；
     * - arrived：已到达，party 到达某一个阶段时的状态；
     * - phase：阶段计数器，初始阶段为 0，最大阶段为 Integer.MAX_VALUE，当 phase 达到 Integer.MAX_VALUE时,会被重置为 0, 开始下一轮循环；
     * - advance：阶段前进点，可以使用 arriveAndAwaitAdvance() [到达并等待前进] 等待进入下一个阶段；也可以使用 onAdvance(int phase, int registeredParties) 选择（返回 false）进入下一个阶段，或（返回 true）终止。
     * - deregister：退出者，一个线程在 arrive 到某个 phase 后可以选择退出同步（类似于参赛者中途退赛），可以使用 arriveAndDeregiter() [到达并退出] 方法来实现。
     * 核心方法：
     * - public Phaser(int parties) -> 指定参与者数量。
     * - protected boolean onAdvance(int phase, int registeredParties) -> 覆写该方法，控制阶段数 phaseNum，实现 Phaser 终止逻辑（phase == phaseNum || registeredParties == 0）。
     * - public int arriveAndAwaitAdvance() -> 到达当前阶段并等待进行下一个阶段（到达数量（-> 参与者计数器） + 1）。
     * - public int arriveAndDeregister() -> 到达当前阶段并退出（参与者计数器 - 1）。
     * 使用步骤：
     * 1）调用构造器，创建 [移相器]，指定 Phaser 的 parties（参与者）数量，例如 2 个;
     * 2）覆写 onAdvance() 方法，控制阶段数 phaseNum，实现 Phaser 终止（phase == phaseNum || registeredParties == 0）逻辑。
     * 3）每当一个线程到达当前阶段（phase 0）时，调用 Phaser.arriveAndAwaitAdvance() 方法，进行自我等待，等待所有线程都运行到（arrive）当前阶段（phase 0）时被唤醒，继续各自前往（advance）下一个阶段（phase 1），依次类推，直到终点。
     * Phaser 与 CyclicBarrier 的异同点：
     * 1. 相同点：二者都用于控制各个并发子任务线程之间的同步关系。
     * 2. 不同点：
     * - CyclicBarrier 可以理解为 [单阶段 - 屏障点] 同步机制。
     * - Phaser 提供了更灵活的 [多阶段] 同步机制，比 CyclicBarrier 强大的多。
     * --> Phaser 可以在每个阶段增加新的参与者，只要 register 即可，也可以选择 deregister 减少参与者；此类方法仅仅是增加 phaser 中参与者的数量。
     * --> Phaser 甚至可以完成类似 Master-Worker 高并发模式的任务。
     */
    public static void main(String[] args) {
        System.out.println("*** [Phaser] 移相器，并发任务线程 [多阶段] 同步控制器 ***");
        System.out.println("描述信息：总共分为 3 个阶段，3 个参与者。");
        System.out.println("---------------------------------------------");

        Phaser phaser = new DemoPhaser(3, 3);

        ExecutorService executorService = Executors.newFixedThreadPool(3);
        for (int i = 0; i < 3; i++) {
            executorService.execute(() -> {
                System.out.println(Thread.currentThread().getName() + " 参与者，正在运行...");
                try {
                    Thread.sleep(ThreadLocalRandom.current().nextInt(3000));
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

                System.out.println(Thread.currentThread().getName() + " 参与者，已到达 [phase 0]...");
                phaser.arriveAndAwaitAdvance();

                System.out.println(Thread.currentThread().getName() + " 参与者，正在运行...");
                try {
                    Thread.sleep(ThreadLocalRandom.current().nextInt(3000));
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

                System.out.println(Thread.currentThread().getName() + " 参与者，已到达 [phase 1]...");
                phaser.arriveAndAwaitAdvance();

                System.out.println(Thread.currentThread().getName() + "参与者，正在运行...");
                try {
                    Thread.sleep(ThreadLocalRandom.current().nextInt(3000));
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

                System.out.println(Thread.currentThread().getName() + "参与者，运行结束！！！");
            });
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        executorService.shutdown();
    }

    /**
     * 继承 Phaser，复写 onAdvance() 方法。
     * - 控制阶段数 phaseNum，实现 Phaser 终止逻辑（phase == phaseNum || registeredParties == 0）。
     */
    private static class DemoPhaser extends Phaser {
        private int phaseNum;

        public DemoPhaser(int parties, int phaseNum) {
            super(parties);
            this.phaseNum = phaseNum;
        }

        /**
         * 核心方法：用于终止 [Phaser] 移相器。
         */
        @Override
        protected boolean onAdvance(int phase, int registeredParties) {
            System.out.println("当前阶段：" + phase + " 已经结束了，参与者数量 -> " + registeredParties + ", 开始各自运行 " + (phase + 1) + " 阶段的任务...");
            return phase == phaseNum || registeredParties == 0;
        }
    }

}
